<?php
namespace Plu\Library\Services;

use Zookeeper,
    RdKafka\Producer as KafkaProducer,
    RdKafka\Consumer as KafkaConsumer,
    RdKafka\TopicConf as TopicConf,
    RdKafka\Conf as KafkaConf;

class KafkaService extends ServiceBase {
    private $topic;

    private $servers;

    private $client = [];

    private $kafkaConfig;

    const CACHE_TIME = 120;

    public function __construct($zookeeker, $path, $topic, $conf = null) {
        $this->topic = $topic;
        $this->servers = $this->getBrokers($zookeeker, $path);

        if (empty($conf) || !($conf instanceof KafkaConf)) {
            $conf = new KafkaConf;

            $conf->set('socket.timeout.ms', 20); // or socket.blocking.max.ms, depending on librdkafka version
            $conf->set('socket.blocking.max.ms', 50);
            $conf->set('socket.blocking.max.ms', 5);
            $conf->set('socket.keepalive.enable', true);
            $conf->set('queue.buffering.max.ms', 5);

            if (function_exists('pcntl_sigprocmask')) {
                pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
                $conf->set('internal.termination.signal', SIGIO);
            }
        }
        $this->kafkaConfig = $conf;
    }

    /**
     * 从Zookeeper获取kafka的配置项
     *
     * $zookeeker  zookeeper服务器
     * $path       配置路径
     */
    public function getBrokers($zookeeker, $path) {
        $guid = sprintf("%s_server_config", $this->topic);
        $servers = $this->cache->get($guid, self::CACHE_TIME);

        if (empty($servers)) {
            $zookeeker = new Zookeeper($zookeeker);

            if (strlen($path) > 1 && preg_match('@/$@', $path)) {
                // remove trailing /
                $path = substr($path, 0, -1);
            }

            $path = $path . '/brokers/ids';
            $servers = [];

            if (!$zookeeker->exists($path)) {
                throw new \Exception(sprintf("Zookeeper path: %s not found.", $path));
            }

            $brokers = $zookeeker->getChildren($path);

            foreach ($brokers as $brokerId) {
                $item = $zookeeker->get($path . '/' . $brokerId);
                $item = json_decode($item, true);
                $servers[] = sprintf("%s:%s", $item['host'], $item['port']);
            }
            $this->cache->save($guid, $servers, self::CACHE_TIME);
        }

        if (!empty($servers) && is_array($servers)) {
            $servers = join(",", $servers);
        }

        return $servers;
    }

    /**
     * 发送kafka消息
     */
    public function produce($msg, $key = null) {
        if (empty($this->servers)) {
            return;
        }
        if (!isset($this->client['producer'])) {
            $kafka = new KafkaProducer($this->kafkaConfig);
            $kafka->addBrokers($this->servers);

            $this->client['producer'] = $kafka;
            $this->client['producer_topic'] = $kafka->newTopic($this->topic);
        }

        if (is_array($msg)) {
            $msg = json_encode($msg);
        }

        $this->client['producer_topic']->produce(RD_KAFKA_PARTITION_UA, 0, $msg, $key);

        while ($this->client['producer']->getOutQLen() > 0) {
            $this->client['producer']->poll(5);
        }

        return true;
    }

    /**
     * 消费kafka消息
     */
    public function consume($offset = RD_KAFKA_OFFSET_BEGINNING, $groupId = "") {
        if (!$this->client['consumer']) {

            if (empty($groupId)) {
                $groupId = sprintf("php-group-%s", $this->topic);
            }
            $this->kafkaConfig->set('group.id', $groupId);
            $this->kafkaConfig->set('offset.store.method', 'broker');

            $kafka = new KafkaConsumer($this->kafkaConfig);
            $kafka->setLogLevel(LOG_DEBUG);
            $kafka->addBrokers($this->servers);

            $queue = $kafka->newQueue();
            if ($offset == RD_KAFKA_OFFSET_STORED) {
                $topicConf = new TopicConf();

                $topicConf->set("auto.commit.enable", true);
                $topicConf->set("auto.commit.interval.ms", 1000);
                $topicConf->set("offset.store.sync.interval.ms", 6000);
                $topicConf->set('offset.store.method', 'broker');
                $topicConf->set('auto.offset.reset', 'largest');

                $topic = $kafka->newTopic($this->topic, $topicConf);
            } else {
                $topic = $kafka->newTopic($this->topic);
            } 

            //getmetadata
            $metadata = $kafka->getMetadata(false, $topic, 60e3);
            $topics = $metadata->getTopics();
            $topicInfo = $topics->current();
            $partitions = $topicInfo->getPartitions();
            
            foreach ($partitions as $partition) {
                $topic->consumeQueueStart($partition->getId(), $offset, $queue);
            }

            $this->client['consumer'] = $queue;
        }

        while (true) {
            $msg = $this->client['consumer']->consume(1000);
            if ($offset == RD_KAFKA_OFFSET_STORED) {
                if ( empty( $msg->err ) && !empty( $msg->payload ) ) {
                    $topic->offsetStore($msg->partition, $msg->offset);
                }
            }

            yield $msg;
        }
    }
}
?>
